use std::{collections::HashMap, ops::DerefMut};

use crate::RedisConnectionManager;
use crate::ManageConnection;
use redis::Commands;
use tokio::task::JoinHandle;
// use tokio_with_wasm::alias as tokio;

/// StdBoxError
pub type StdBoxError = Box<dyn std::error::Error + Send + Sync>;
/// IResult
pub type IResult<V = ()> = Result<V, StdBoxError>;


/// RedisPool
pub struct RedisPool{
    /// 多线程异步运行时
    rt: tokio::runtime::Runtime,
    /// redis地址
    redis_addr: String,
    /// 为提高性能，要求使用者必须提前指定
    db_nums: Vec<u8>,
    map_pool: HashMap<u8, crate::Pool<RedisConnectionManager>>
}

impl RedisPool {
    /// 为提高性能，要求使用者必须提前指定db_nums
    pub fn new(redis_addr: String, db_nums: Vec<u8>) -> IResult<Self>{
        // 1. generate map_pool =================================
        let mut map_pool = HashMap::new();
        for db_num in db_nums.iter() {
            let addr = format!("{redis_addr}{db_num}");
            // 1.1 manager
            let manager = RedisConnectionManager::new(addr).map_err(|e| format!("redis::RedisError: {:?}", e))?;
            // 2. connection
            {
                let mut conn = manager.connect().expect("r2d2_redis connect()");
                manager
                    .is_valid(&mut conn).map_err(|e| format!("r2d2_redis::Error: {:?}", e))?;
            }
            // 3. pool
            let pool = crate::Pool::builder()
                .build(manager)?;
            map_pool.insert(*db_num, pool);
        }
        // 2. generate map_pool =================================
        let rt = tokio::runtime::Builder::new_current_thread()
                        .enable_all()
                        .worker_threads(4)
                        .build()?;
        

        Ok(Self{
            rt,
            redis_addr,
            db_nums,
            map_pool,
        })
    }


    fn spawn_blocking<F, R>(&self, db_num: u8, func: F) -> JoinHandle<IResult<R>>
    where
        F: FnOnce(crate::PooledConnection<crate::RedisConnectionManager>) -> Result<R, redis::RedisError>
            + Send
            + 'static,
        R: Send + 'static,
    {
        let map_pool_c = self.map_pool.clone();
        self.rt.spawn_blocking(move || {
            // let mut conn = self_c.get_conn(db_num);
            let conn = Self::get_conn_inner(&map_pool_c, db_num);
            let conn = match conn{
                Ok(v) => v,
                Err(e) => {
                    error!("spawn_blocking({db_num}), func: {e}");
                    return Err(e);
                }
            };
            let r = func(conn);
            if let Err(e) = &r{
                error!("spawn_blocking({db_num}), func: {e}");
            }
            let r = r?;
            Ok(r)
        })
    }
    /// inner static function for: fn spawn_blocking
    fn get_conn_inner(map_pool: &HashMap<u8, crate::Pool<crate::RedisConnectionManager>>, db_num: u8) -> IResult<crate::PooledConnection<crate::RedisConnectionManager>> {
        
        let pool = match map_pool.get(&db_num) {
            Some(v) => v,
            None => return Err(format!("redis is not initialized at db: {db_num}"))?,
        };

        let conn = pool.get().map_err(|e| format!("r2d2_redis::Error: {:?}", e))?;
        
        Ok(conn)
    }

    pub(crate) fn get_conn(&self, db_num: u8) -> IResult<crate::PooledConnection<crate::RedisConnectionManager>> {
        Self::get_conn_inner(&self.map_pool, db_num)
    }
    /// ping
    pub fn ping(&self) {
        let db_num = self.db_nums[0];
        let mut conn = match self.get_conn(db_num){
            Ok(v) => v,
            Err(e) => {
                println!("RedisPool.ping {} connection error: {:?}", self.redis_addr, e);
                return;
            },
        };
        match redis::cmd("PING")
            .query::<String>(conn.deref_mut()){
                Ok(pong) => println!("RedisPool.ping successs: you send 'PING' and then receive '{pong}'."),
                Err(e) => println!("RedisPool.ping {} query error: {:?}", self.redis_addr, e),
        }
        
    }

    /// keys of db_num
    pub fn keys(&self, db_num: u8, key_pattern: &str) -> IResult<Vec<String>> {
        let mut conn = self.get_conn(db_num)?;    
        let mut keys: Vec<String> = conn.keys(key_pattern)?;
        keys.sort();
        Ok(keys)

    }

    /// h_fields
    pub fn h_fields(&self, db_num: u8, key: &str) -> IResult<Vec<String>>{
        let mut conn = self.get_conn(db_num)?;    
        let mut fields: Vec<String> = conn.hkeys(key)?;
        fields.sort();
        Ok(fields)
    }

    /// hget1_string
    pub fn hget1_string(&self, db_num: u8, key: &str, field: &str) -> IResult<String>{
        let mut conn = self.get_conn(db_num)?;    
        let value: String = conn.hget(key, field)?;
        
        Ok(value)
    }

    /// hget1_bytes
    pub fn hget1_bytes(&self, db_num: u8, key: &str, field: &str) -> IResult<Vec<u8>>{
        let mut conn = self.get_conn(db_num)?;    
        let value: Vec<u8> = conn.hget(key, field)?;
        
        Ok(value)
    }

    /// hget_all_string
    pub fn hget_all_string(&self, db_num: u8, key: &str) -> IResult<HashMap<String, String>> {
        let mut conn = self.get_conn(db_num)?;
        let map_field_value: HashMap<String,String> = conn.hgetall(key)?;
        
        Ok(map_field_value)
    }

    /// hget_all_bytes
    pub fn hget_all_bytes(&self, db_num: u8, key: &str) -> IResult<HashMap<String, Vec<u8>>> {
        let mut conn = self.get_conn(db_num)?;
        let map_field_value: HashMap<String, Vec<u8>> = conn.hgetall(key)?;
        
        Ok(map_field_value)
    }
    
    /// hmget_string
    pub fn hmget_string(&self, db_num: u8, key: &str, fields: &Vec<String>) -> IResult<Vec<String>> {
        let mut conn = self.get_conn(db_num)?;
        let list_value: Vec<String> = conn.hget(key, fields.as_slice())?;
        
        Ok(list_value)
    }
    
    /// hmget_bytes
    pub fn hmget_bytes(&self, db_num: u8, key: &str, fields: &Vec<String>) -> IResult<Vec<Vec<u8>>> {
        let mut conn = self.get_conn(db_num)?;
        let list_value: Vec<Vec<u8>> = conn.hget(key, fields.as_slice())?;
        
        Ok(list_value)
    }

    /// hget_pairs
    pub fn hget_pairs(&self, db_num: u8, pairs_key_field: &Vec<(String, String)>) -> IResult<Vec<String>> {
        let mut conn = self.get_conn(db_num)?;
         let mut pipe = redis::pipe();
         for (key, field) in pairs_key_field{
            pipe.cmd("hget").arg(key).arg(field);
         }

         let list_value = pipe.query::<Vec<String>>(conn.deref_mut())?;
         Ok(list_value)
    }

    /// hmset_pairs
    pub fn hmset_pairs(&self, db_num: u8, key: String, list_pairs: Vec<(String, String)>) -> IResult {
        let mut conn = self.get_conn(db_num)?;
        let mut pipe = redis::pipe();
        pipe.cmd("hmset").arg(key);
        for (field, value) in list_pairs{
            pipe.arg(field).arg(value);
        }

        pipe.query(conn.deref_mut())?;
        Ok(())
    }

    /// hmset_triples
    pub fn hmset_triples(&self, db_num: u8, list_triples: Vec<(String, String, String)>) -> IResult {
        let mut conn = self.get_conn(db_num)?;
        let mut pipe = redis::pipe();
        for (key, field, value) in list_triples{
            pipe.cmd("hset").arg(key).arg(field).arg(value);
        }

        pipe.query(conn.deref_mut())?;
        Ok(())
    }

    /// hmset_triples_async
    pub fn hmset_triples_async(&self, db_num: u8, list_triples: Vec<(String, String, String)>) {
        self.spawn_blocking(
            db_num,
            move |mut conn| {
                let mut pipe = redis::pipe();
                for (key, field, value) in list_triples{
                    pipe.cmd("hset").arg(key).arg(field).arg(value);
                }
        
                pipe.query(conn.deref_mut())?;
            Ok(())
        });
    }


    /// flush_db_async
    pub fn flush_db_async(&self, db_num: u8) {
        self.spawn_blocking(
            db_num,
            move |mut conn| {
                if let Err(e) = redis::cmd("FLUSHDB")
                    .exec(conn.deref_mut()) {
                        error!("RedisPool.flush_db_async {db_num}: {:?}", e);
                    };
                Ok(())
        });
    }

    /// hset1_string_async
    pub fn hset1_string_async(&self, db_num: u8, key: String, field: String, value: String)  {
        self.spawn_blocking(
            db_num,
            move |mut conn| {
                if let Err(e) = conn.hset::<_, _, _, i32>(&key, &field, &value) {
                    error!("RedisPool.hset1_string_sync {key}.{field}: {:?}", e);
                }
                Ok(())
            },
        );


    }

    /// hset1_bytes_async
    pub fn hset1_bytes_async(&self, db_num: u8, key: String, field: String, value: Vec<u8>){
        self.spawn_blocking(
            db_num,
            move | mut conn| {
                if let Err(e) = conn.hset::<_, _, _, i32>(key.as_bytes(), field.as_bytes(), value) {
                    error!("RedisPool.hset1_bytes_async {key}.{field}: {:?}", e);
                }
                Ok(())
            },
        );
    }

    /// hmset_bytes_async
    pub fn hmset_bytes_async(&self, db_num: u8, key: String, list_field_value: Vec<(String, Vec<u8>)>) {
        self.spawn_blocking(
            db_num,
            move | mut conn| {
                let mut list_bytes = Vec::with_capacity(list_field_value.len());
                for (field, value) in list_field_value{
                    list_bytes.push((field.into_bytes(), value));
                }
                if let Err(e) = conn.hset_multiple::<_,_, _, i32>(&key, &list_bytes) {
                    error!("RedisPool.hmset_bytes_async {key}: {:?}", e);
                }
                
                Ok(())
            },
        );
    }
}

#[test]
fn test_20230315() {
    // let redis_地址_测试环境 = "redis://:prod-sz-xtp-61e670fa-48be-4b4b-9786-e39b6e8a06a7@192.168.1.11:63790/";
    let redis_地址_测试环境 = "redis://:dev-22-e0cb756e-f722-48e9-a3de-81cb83789790@222.219.145.9:26379/";
    let redis_地址_测试环境 = "redis://192.168.1.66:6379/";
    
    // 为追求性能，必须提前指定需要访问的 db_nums
    let redis_pool = match RedisPool::new(redis_地址_测试环境.to_string(), vec![1, 2, 3, 4]){
        Ok(v) => v,
        Err(e) => {
            println!("RedisPool初始化失败！{:?}", e);
            return;
        },
    };

    // 该封装里面有println语句
    redis_pool.ping();

    println!("Redis的异步(*_async)函数封装过程中, 使用了error!宏, 若需要在控制台或文件中输出, 请自行指定！");

    let db_num = 1; // 在指定的db_nums里面
    redis_pool.hset1_string_async(db_num, "key_1".to_string(), "field_1".to_string(), "value_1".to_string());

    let db_num = 5; // 不在指定的db_nums里面
    redis_pool.hset1_string_async(db_num, "key_1".to_string(), "field_1".to_string(), "value_1".to_string());

    {
        
        let db_num = 2;
        let key = "20230322-沪深A股_昨日收盘";
        let list_s8证券代码 = vec!["SZ300901".to_string(), "SH688577".to_string(), "SZ002491".to_string(), "SH603529".to_string(), "".to_string()];
        let list_沪深A股 = match redis_pool.hmget_string(db_num, key, &list_s8证券代码){
            Ok(v) => v,
            Err(e) => panic!("hmget_string error: {:?}", e),
        };
        println!("\t{} 条沪深A股: {:?}", list_沪深A股.len(), list_沪深A股);

        let db_num = 3;
        let key = "20230324-深度行情简易";
        let list_深度行情简易 = match redis_pool.hmget_bytes(db_num, key, &list_s8证券代码){
            Ok(v) => v,
            Err(e) => panic!("hmget_bytes error: {:?}", e),
        };
        println!("\t{} 条深度行情简易", list_深度行情简易.len());
    
    }

    {
        // 你也可以自己获取连接，进行其他未封装的Redis操作
        let db_num = 1;
        let mut conn = match redis_pool.get_conn(db_num) {
            Ok(v) => v,
            Err(e) => panic!("Redis_Pool get db {db_num} connection error: {:?}", e),
        };
        
        match redis::cmd("PING")
            .query::<String>(conn.deref_mut()){
                Ok(pong) => println!("RedisPool.ping successs: you send 'PING' and then receive '{pong}'."),
                Err(e) => println!("RedisPool.ping query error: {:?}", e),
        }
    }
    

    // println!("{}", hget1_string(2, "20230320-沪深A股".to_string(), "SZ002949".to_string()).unwrap_or_else(|| "20230315-沪深A股::SZ002949".to_string()));
    redis_pool.hset1_string_async(1, "hello1".to_string(), "world1".to_string(), "这是一个无聊的字符串".to_string());
    redis_pool.hset1_bytes_async(3, "hello3".to_string(), "world3".to_string(), vec![0, 0, 0, 0]);

    let mut list_field_value = vec![];
    list_field_value.push(("fieldA".to_string(), "valuaA".to_string()));
    list_field_value.push(("fieldB".to_string(), "valuaB".to_string()));
    list_field_value.push(("fieldA".to_string(), "valuaC".to_string()));
    _ = redis_pool.hmset_pairs(4, "hello5".to_string(), list_field_value);

    redis_pool.flush_db_async(4);




    // std::thread::sleep(sdt::time::Duration::from_secs(60));

}


#[test]
fn test_20230410() {
    // let redis_地址_测试环境 = "redis://:prod-sz-redis-61e670fa-48be-4b4b-9786-e39b6e8a06a7@192.168.1.11:26379/";
    let redis_地址_测试环境 = "redis://:prod-redis-11-bea8a019-49fc-4b5d-bfef-e9fcaa92683b@192.168.1.11:6379/";
    let redis_地址_测试环境 = "redis://192.168.1.66:6379/";
    
    // 为追求性能，必须提前指定需要访问的 db_nums
    let redis_pool = match RedisPool::new(redis_地址_测试环境.to_string(), vec![1, 2, 3, 4]){
        Ok(v) => v,
        Err(e) => {
            println!("RedisPool初始化失败！{:?}", e);
            return;
        },
    };
    
    let mut list_triples = vec![];
    list_triples.push(("key1".to_string(), "1".to_string(), "1".to_string()));
    list_triples.push(("key2".to_string(), "1".to_string(), "1".to_string()));
    list_triples.push(("key3".to_string(), "1".to_string(), "1".to_string()));
    list_triples.push(("key4".to_string(), "1".to_string(), "1".to_string()));
    let count = list_triples.len();
    match redis_pool.hmset_triples(4, list_triples){
        Ok(_) => println!("list_triples: hset {count} records." ),
        Err(e) => println!("hset_pairs error: {:?}.", e),
    }

    let mut list_pairs = vec![];
    list_pairs.push(("field1".to_string(), "1".to_string()));
    list_pairs.push(("field2".to_string(), "1".to_string()));
    list_pairs.push(("field3".to_string(), "1".to_string()));
    list_pairs.push(("field4".to_string(), "1".to_string()));
    let count = list_pairs.len();
    match redis_pool.hmset_pairs(4, "key_pairs".to_string(), list_pairs) {
        Ok(_) => println!("hmset_pairs: hset {count} records." ),
        Err(e) => println!("hset_pairs error: {:?}.", e),
    }
    
    // std::thread::sleep(Duration::from_secs(60));
}
